package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.base.VisitorStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashMap;

import static com.atguigu.gmall.realtime.common.Constant.*;

/**
 * @author liuyun
 * @title: VisitorStatsApp
 * @projectName gmal0924
 * @description: TODO
 * @date 2022-3-25 11:44
 */
public class VisitorStatsApp extends BaseAppV2 {

    public static void main(String[] args) {
        new VisitorStatsApp().init(4001,
                1,
                "VisitorStatsApp",
                "VisitorStatsApp",
                TOPIC_DWD_PAGE, TOPIC_DWM_UV, TOPIC_DWM_UJ
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> topicStreamMap) {
        //1、多个流union成一个流
//        topicStreamMap.get(TOPIC_DWD_PAGE).print("page");
//        topicStreamMap.get(TOPIC_DWM_UV).print("uv");
//        topicStreamMap.get(TOPIC_DWM_UJ).print("uj");

        DataStream<VisitorStats> visitorStatsDataStream = unionOne(topicStreamMap);
        //visitorStatsDataStream.print();
        //2、开窗聚合
        SingleOutputStreamOperator<VisitorStats> visitorStatsSingleOutputStream = windowAndAgg(visitorStatsDataStream);

        //visitorStatsSingleOutputStream.print();
        //3、数据写入到岗clickhouse中

        writeToClickHouse(visitorStatsSingleOutputStream);


    }

    private void writeToClickHouse(SingleOutputStreamOperator<VisitorStats> visitorStatsSingleOutputStream) {
        visitorStatsSingleOutputStream
                .addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB,
                                Constant.CLICKHOUSE_TABLE_VISITOR_STATS_2022,
                                VisitorStats.class
                        )
                );
    }

    private SingleOutputStreamOperator<VisitorStats> windowAndAgg(DataStream<VisitorStats> visitorStatsDataStream) {
        return visitorStatsDataStream.
                assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                                .withTimestampAssigner((vs, ts) -> vs.getTs())
                ).keyBy(vs -> vs.getVc() + "_" + vs.getCh() + "_" + vs.getAr() + "_" + vs.getIs_new())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<VisitorStats>("late") {
                })
                .reduce(new ReduceFunction<VisitorStats>() {
                            @Override
                            public VisitorStats reduce(VisitorStats value1,
                                                       VisitorStats value2) throws Exception {
                                value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                                value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                                value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                                value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>.Context context,
                                                Iterable<VisitorStats> elements,
                                                Collector<VisitorStats> out) throws Exception {
                                VisitorStats vs = elements.iterator().next();
                                String stt = AtguiguUtil.toDateTime(context.window().getStart());
                                String edt = AtguiguUtil.toDateTime(context.window().getEnd());

                                vs.setStt(stt);
                                vs.setEdt(edt);
                                vs.setTs(System.currentTimeMillis());  // 改成统计时间
                                out.collect(vs);
                            }
                        }
                );
    }

    private DataStream<VisitorStats> unionOne(HashMap<String, DataStreamSource<String>> topicStreamMap) {
        //pv
        SingleOutputStreamOperator<VisitorStats> pvAndDuringTimeStream = topicStreamMap
                .get(TOPIC_DWD_PAGE)
                .map(json -> {
                    JSONObject jsonObject = JSON.parseObject(json);
                    JSONObject common = jsonObject.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String isNew = common.getString("is_new");

                    Long ts = jsonObject.getLong("ts");

                    Long duringTime = jsonObject.getJSONObject("page").getLong("during_time");

                    return new VisitorStats(
                            "", "",
                            vc, ch, ar, isNew,
                            0L, 1L, 0L, 0L, duringTime,
                            ts
                    );
                });
        //uv
        SingleOutputStreamOperator<VisitorStats> uvStream = topicStreamMap
                .get(TOPIC_DWD_PAGE)
                .map(json -> {
                    JSONObject jsonObject = JSON.parseObject(json);
                    JSONObject common = jsonObject.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String isNew = common.getString("is_new");

                    Long ts = jsonObject.getLong("ts");

                    return new VisitorStats(
                            "", "",
                            vc, ch, ar, isNew,
                            1L, 0L, 0L, 0L, 0L,
                            ts
                    );
                });
        //3、uj

        SingleOutputStreamOperator<VisitorStats> ujStream = topicStreamMap
                .get(TOPIC_DWD_PAGE)
                .map(json -> {
                    JSONObject jsonObject = JSON.parseObject(json);
                    JSONObject common = jsonObject.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String isNew = common.getString("is_new");

                    Long ts = jsonObject.getLong("ts");

                    return new VisitorStats(
                            "", "",
                            vc, ch, ar, isNew,
                            1L, 0L, 0L, 1L, 0L,
                            ts
                    );
                });
        //4、进入 来源topic怎么判断sv
        SingleOutputStreamOperator<VisitorStats> svStream = topicStreamMap
                .get(TOPIC_DWM_UJ)
                .flatMap(new FlatMapFunction<String, VisitorStats>() {
                    @Override
                    public void flatMap(String json, Collector<VisitorStats> out) throws Exception {

                        JSONObject jsonObject = JSON.parseObject(json);
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        boolean b = lastPageId == null || lastPageId.length() == 0;
                        if (lastPageId == null || lastPageId.length() == 0) {
                            JSONObject common = jsonObject.getJSONObject("common");
                            String vc = common.getString("vc");
                            String ch = common.getString("ch");
                            String ar = common.getString("ar");
                            String isNew = common.getString("is_new");

                            Long ts = jsonObject.getLong("ts");

                            VisitorStats vs = new VisitorStats(
                                    "", "",
                                    vc, ch, ar, isNew,
                                    0L, 0L, 1L, 0L, 0L,
                                    ts
                            );
                            out.collect(vs);
                        }

                    }
                });
        return pvAndDuringTimeStream.union(uvStream, ujStream, svStream);
    }

}
